package ckR

import (
	metaD "db2s/MetaInit"
	"db2s/arg"
	cp "db2s/componentInit"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/pubM"
	quit "db2s/quitDispos"
	"db2s/ref"
	mq "db2s/topic-mq"
	"errors"
	"fmt"
	"github.com/panjf2000/ants/v2"
	"github.com/vbauerster/mpb/v8"
	"os"
	"reflect"
	"runtime/pprof"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var monitorListener mq.MonitorListener

type sendTopicF func(check CheckMode, topic Topic, tableName string, msg map[string]any) bool
type sendCallFuncAction func(s any, d int64)

func GetCheckSchemaMetaData() *CheckSchemaMetaData {
	return &CheckSchemaMetaData{TableObjectFlow: make(chan any, global.ChannelDefault)}
}
func (t CheckSchemaMetaData) ChanSchemaContextObjectInit() *SchemaInitContext {
	var tableFree, tableAccount, breakPointSum int64
	return &SchemaInitContext{ChanSchemaContext: make(chan any, global.ChannelDefault), TableSum: t.ObjectSum, IgnoreObjectSum: t.IgnoreObjectSum, TableAccount: &tableAccount, TableFree: &tableFree, BreakPointSum: &breakPointSum}
}
func newSchemaInitContext(seq int, v arg.InputTableAttributes, logSeq int64) *SchemaInitContext {
	return &SchemaInitContext{
		Schema: v.Schema, Table: v.Table, MSchema: v.MSchema,
		MTable: v.MTable, Seq: seq, ObjectLogSeq: logSeq,
		Role: v.TableAttributes.Role, TableAttributes: v.TableAttributes,
	}
}
func (t CheckSchemaMetaData) objectMetaInitAbnormalQuit(seq int, v arg.InputTableAttributes, res *SchemaInitContext) bool {
	var err error
	if res.TPod, err = t.podInit(seq, v); err != nil {
		return true
	}
	if atomic.LoadInt64(v.TableStatus) == -1 {
		res.TPod.TableMiss = true
		return true
	}
	return false
}
func (t CheckSchemaMetaData) objectMetaInitArgs(v arg.InputTableAttributes, re *SchemaInitContext) (res []reflect.Value) {
	return []reflect.Value{
		reflect.ValueOf(metaD.ObjectMetaInput{
			TableAttributes: v,
			TIndex:          re.IndexData,
		}),
	}
}
func (t CheckSchemaMetaData) objectMetaResult(logSeq int64, p string, v arg.InputTableAttributes, in []reflect.Value, res *SchemaInitContext) (err error) {
	var event = "[objectMetaInit]"
	for y, x := range in {
		switch y {
		case 0:
			switch p {
			case "struct":
				res.TColumnMeta = x.Interface().(global.TableMetaGather)
			case "index":
				res.IndexData = x.Interface().(metaD.IndexChooseResult)
			case "part":
				res.Meta.Part = x.Interface().(global.PartMetaMarge)
			}
			res.TPod.IndexType = res.IndexData.IndexType
			res.TPod.IndexCol = strings.Join(res.IndexData.IndexColumn, ",")
		case 1:
			if x.Interface() != nil {
				err = x.Interface().(error)
				errorWriteInfo(fmt.Sprintf("(%v) %v An error occurred in obtaining the table %v of table %v.%v and will be skipped. error is %v", logSeq, event, p, v.Schema, v.Table, err))
				return
			}
		}
	}
	return
}
func getObjectMetaFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		//"struct": reflect.ValueOf(getColMe),
		//"index":  reflect.ValueOf(getIndexMe),
		//"part":   reflect.ValueOf(getPartMe),
	}
}
func (t CheckSchemaMetaData) objectMetaInit(seq int, v arg.InputTableAttributes, logSeq int64) (res *SchemaInitContext, err error) {
	//var v1 = initMetaD(logSeq)
	res = newSchemaInitContext(seq, v, logSeq)
	//if t.objectMetaInitAbnormalQuit(seq, v, res) {
	//	res.TableMiss = true
	//	return
	//}
	//for _, p := range ref.GetJSONFieldNames(res.BaseMe) {
	//	fun, ok := getObjectMetaFunc()[p]
	//	if !ok {
	//		continue
	//	}
	//	if err = t.objectMetaResult(logSeq, p, v, fun.Call(reflectValue(v)), res); err != nil {
	//		err = ref.ErrAddPrintf("objectMetaInit", err)
	//		return
	//	}
	//}

	//for _, p := range []string{"struct"} {
	//	var fun reflect.Value
	//	var ok bool
	//	if fun, ok = getObjectMetaFunc(v1)[p]; !ok {
	//		continue
	//	}
	//	if err = t.objectMetaResult(logSeq, p, v, fun.Call(t.objectMetaInitArgs(v, res)), res); err != nil {
	//		fmt.Println("------err：", err)
	//		return
	//	}
	//}
	return
}

func (t CheckSchemaMetaData) schemaMetaExecFunc(event string, f any) (r any) {
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	baseParameter, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	seq := baseParameter["seq"].(int)
	logSeq := baseParameter["logSeq"].(int64)
	if p, err := t.objectMetaInit(seq, chanParameter.(arg.InputTableAttributes), logSeq); err != nil {
		log.ErrorLog().Error(fmt.Sprintf("%v", err))
		return
	} else {
		ChanSchemaContext := userParameter[0].(*SchemaInitContext)
		atomic.AddInt64(ChanSchemaContext.TableAccount, 1)
		atomic.AddInt64(ChanSchemaContext.TableFree, 1)
		ChanSchemaContext.ChanSchemaContext <- p
	}
	return
}
func schemaMetaExecFuncParameterSplit(f []any) (baseParameter map[string]any, chanParameter any, useParameter []any) {
	baseParameter = f[0].(map[string]any)
	chanParameter = f[1]
	useParameter = f[2:]
	return
}

func (t CheckSchemaMetaData) Check() {
	continueTableList = breakP.GetRes("result").(map[string]int)
	ChanSchemaContext := t.ChanSchemaContextObjectInit()
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init per table object group metadata -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	go func() {
		defer func() {
			close(ChanSchemaContext.ChanSchemaContext)
		}()
		if err := pubM.ObjectMetaDataInit(pubM.ObjectMetaDataPoint{
			NewBreakPoint: pubM.NewBreakPoint{
				BreakPointSum:     ChanSchemaContext.BreakPointSum,
				ContinueTableList: continueTableList,
				ObjectFlow:        t.TableObjectFlow,
				TableAccount:      ChanSchemaContext.TableAccount,
				TableFree:         ChanSchemaContext.TableFree,
			},
			ExecFunc: global.EFunc{FuncName: t.schemaMetaExecFunc, Params: []any{ChanSchemaContext}},
		}).NormalChanSendMsg("ckR/Check"); err != nil {
			return
		}
	}()
	global.GetIoc().RegisterContext("schemaInitContext", ChanSchemaContext)
}

func FuncExportVar() {
	//函数调用映射
	FuncRef = global.GetIoc().GetBean("checkObjectFuncContext").(map[string]reflect.Value)
	ScheduleTaskInitFuncRef = global.GetIoc().GetBean("scheduleTaskInitFuncContext").(map[string]reflect.Value)
	closeObject = global.GetIoc().GetBean("closeObject").(quit.Close)
	MqListenerSInitFunc = global.GetIoc().GetBean("mqListenerSInitFuncContext").(map[string]reflect.Value)
}

// ExportVar 导出变量 方便使用
func ExportVar() {
	//producer 生产者
	producer = global.GetIoc().GetBean("producer").(mq.Producer)
	//consumer 消费者
	consumer = global.GetIoc().GetBean("consumer").(mq.Consumer)
	//monitorListener mq 监控
	monitorListener = global.GetIoc().GetBean("monitorListener").(mq.MonitorListener)
	PlanContext = global.GetIoc().GetBean("schedulePlanContext").(*SchedulePlanContext) //注册ioc 容器
}

func (context *SchedulePlanContext) GetMode() CheckMode {
	c := global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig)
	switch c.rules.CheckMode {
	case "rows":
		return "rows"
	}
	return "count"
}
func getCheckMod() CheckMode {
	c := global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig)
	switch c.rules.CheckMode {
	case "rows":
		return PlanContext.GetMode()
	}
	return ""
}

func (context *SchedulePlanContext) objectEndMark(index CheckMode) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].ObjectEndMark()
}

var normalSum int

func rowsToTopic(plan *SchedulePlan2) {
	var (
		topic Topic
		table = fmt.Sprintf("%v.%v", plan.Object.Schema, plan.Object.Table)
	)
	normalSum++
	topic = IsMode
	sendMsg(getCheckMod(), topic, table, map[string]any{"segmentSql": map[string]string{}, "plan": plan, "subTask": plan.subTaskInfo, "topic": string(topic)})
}

func checkObjectFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"rows": reflect.ValueOf(rowsToTopic),
	}
}
func sendScheduleMsgFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"rows": reflect.ValueOf(initRowsSchedulePlan),
	}
}
func mqListenerSInitFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"rows": reflect.ValueOf(rowsMQListenInitFunc),
	}
}
func schedulePlanGetIocInit() {
	global.GetIoc().RegisterContext("checkObjectFuncContext", checkObjectFunc())
	global.GetIoc().RegisterContext("scheduleTaskInitFuncContext", sendScheduleMsgFunc())
	global.GetIoc().RegisterContext("mqListenerSInitFuncContext", mqListenerSInitFunc())
	FuncExportVar()
}
func schedulePlanThreadPoolInit() (pool *ants.Pool, err error) {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	//初始化上下文线程池
	if pool, err = ants.NewPool(initContext.TableSum); err != nil {
		return
	}
	progress := mpb.New(mpb.WithWidth(80))
	global.GetIoc().RegisterContext("progress", progress)
	return
}
func planContextBarInit(config *SchedulePlanContextConfig) *mq.TaskBar {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	return cp.InitBarMonitor(cp.Bar{
		TaskTotal:     initContext.TableAccount,
		Task:          initContext.TableSum,
		Free:          initContext.TableFree,
		BreakPointSum: initContext.BreakPointSum,
		CheckMod:      config.GtCheckConfig.SecondaryL.RulesV.CheckMode,
		Object:        config.GtCheckConfig.SecondaryL.SchemaV.CheckIndexObject,
		BeginTime:     config.BeginTime,
		TtyBarFin:     GlobalPConfigs.FilePoint.TtyBarFin,
	}, len(GetCheckToTopicMQListeners()[cp.CheckMode(config.GtCheckConfig.SecondaryL.RulesV.CheckMode)]))
}
func PlanContextInit(config *SchedulePlanContextConfig) (PlanContext *SchedulePlanContext, err error) {
	var pool *ants.Pool
	//初始化上下文
	if pool, err = schedulePlanThreadPoolInit(); err != nil {
		return
	}
	PlanContext = &SchedulePlanContext{
		Pool:                      pool,
		AutoCloseMqComponentTable: cp.InitAutoCloseMqComponentTable(),
		BeginTime:                 config.BeginTime,
	}
	PlanContext.TaskBarTable = planContextBarInit(config) //初始化执行计划表
	PlanContext.ChanSchedulePlanTable = InitToSchedulePlan2Table()
	global.GetIoc().RegisterContext("schedulePlanContext", PlanContext) //注册ioc 容器
	return
}

// NewSchedulePlanContext 创建执行计划上下文
func schedulePlanContext(config *SchedulePlanContextConfig) (PlanContext1 *SchedulePlanContext) {
	var err error
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init schedule task Connext -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	lock.Lock()
	defer lock.Unlock()
	schedulePlanGetIocInit()
	if err = initContextBean(config); err != nil {
		fmt.Println("创建执行计划上下文 NewSchedulePlanContext 失败", err)
		return nil
	}
	if PlanContext1, err = PlanContextInit(config); err != nil {
		return nil
	}
	//导出变量
	ExportVar()
	return
}

// initContextBean 初始化上下文需要用到的组件
func initContextBean(config *SchedulePlanContextConfig) error {
	//TODO: checkConfig 行哥 你来写
	createMq := config.MQConfig.CreateMq(GetCheckToTopicMQListeners())
	if createMq == nil {
		return errors.New("初始化 mq 失败")
	}
	return nil
}

func (context *SchedulePlanContext) Close() {
	for _, component := range context.AutoCloseMqComponentTable {
		context.wg.Add(1)
		func() {
			defer context.wg.Done()
			component.Close(false)
		}()
	}
	context.wg.Wait()
}
func missTableAction(k *SchedulePlan2) {
	event := "[miss]"
	infoWriteInfo(fmt.Sprintf("(%v) %v Table %v.%v does not exist on the target side, and the table status is {missTable}", k.Object.ObjectLogSeq, event, k.Object.Schema, k.Object.Table))
	newTer.Write(k.TPod)
	k.XlsMiss.Data.Missing = GlobalPConfigs.dSns.DestDBName
	newCell.Write(*k.XlsMiss)
	PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "finish")
}

func resultAsyncActive(wg *sync.WaitGroup) {
	var (
		activeDataFileTailMap = map[string]any{
			"count": global.XCountSheetTail{},
			"rows":  global.XRowsSheetTail{},
			"samp":  global.XSampsheetTatil{},
		}
		closeLevel bool
	)
	defer func() {
		wg.Done()
	}()
	var l []string
	for _, v := range global.StructNameJsonToSlice(activeDataFileTailMap[GlobalPConfigs.rules.CheckMode]) {
		l = append(l, fmt.Sprintf("%v", v))
	}
	a := fmt.Sprintf("%v", strings.Join(l, " || "))
	if err := writeBreakPointData(GlobalPConfigs.FilePoint.ActiveDataFin, []*string{&a}, 1); err != nil {
		//return
	}
	for {
		select {
		case m, ok := <-ResultStatisticsMq:
			var (
				sheetName string
				data      any
				xlsData   map[string]any
			)
			if !ok {
				if err := ResultStatisticsGeneral.Close(); err != nil {
					return
				}
				return
			}
			if v, ok1 := m.(map[string]any)["sheet"]; ok1 {
				sheetName = v.(string)
			}
			if v, ok1 := m.(map[string]any)["data"]; ok1 {
				if reflect.TypeOf(v).Kind() == reflect.Struct {
					data = strings.Join(global.StructJsonPointerValueToSlice(v), global.StructColumnSplit)
				} else {
					data = v
				}
			}
			xlsData = map[string]any{
				"sheet": sheetName,
				"data":  data,
			}
			if !ResultStatisticsGeneral.ResultAsynchronousWriteFile(xlsData) {
				return
			}
		case c, ok := <-BarResultStatisticMq:
			if ok && c != nil {
				var activeDataString string
				if reflect.TypeOf(c).Kind() == reflect.Struct {
					activeDataString = strings.Join(global.StructJsonPointerValueToSlice(c), " || ")
				} else {
					activeDataString = fmt.Sprintf("%v", c)
					activeDataString = strings.ReplaceAll(activeDataString, "\n", ",")
				}
				if err := writeBreakPointData(GlobalPConfigs.FilePoint.ActiveDataFin, []*string{&activeDataString}, 1); err != nil {
					//return
				}
				var sheetName string
				for k, v := range GlobalPConfigs.result.XlsV.XlsSheetStructMap {
					if reflect.TypeOf(c) == v {
						sheetName = k
						break
					}
				}
				ResultStatisticsMq <- map[string]any{"sheet": sheetName, "data": c}
			} else {
				if !closeLevel {
					close(ResultStatisticsMq)
					closeLevel = true
				}
			}
		}
	}
}
func (context *SchedulePlanContext) monitor() {
	switch GlobalPConfigs.result.Teletypewriter {
	case "bar":
		context.TaskBarTable.BarMonitor()
	case "topic":
		monitorListener.TurnMonitor()
	default:
		context.TaskBarTable.BarMonitor()
	}
}
func sendStartMsgContinueFunc(event string, f any) (r any) {
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	k := chanParameter.(*SchedulePlan2)
	if k.TPod.TableMiss {
		missTableAction(k)
		return true
	}
	return tableObjectErrorStatus(k)
}
func sendStartMsgExecFunc(event string, f any) (r any) {
	global.GTableCurry <- struct{}{}
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	if funcName, ok := FuncRef[GlobalPConfigs.rules.CheckMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{reflect.ValueOf(chanParameter)}
			funcName.Call(arguments)
		}
	} else {
		fmt.Println(fmt.Sprintf("-- [%v] %v report: checkObject parameter selection error, please refer to the log file for details, enable debug to get more information -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		os.Exit(1)
	}
	return
}
func sendStartMsgFinishFunc(event string, _ any) (r any) {
	global.MyWait.Done()
	PlanContext.objectEndMark(getCheckMod())
	return
}
func tableObjectErrorStatus(k *SchedulePlan2) bool {
	if atomic.LoadInt64(k.Status.ErrEnd) >= 0 {
		return false
	}
	newTer.Write(k.TPod)
	newCell.Write(k.XlsTerminal)
	newCell.Write(k.XlsResult)
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "error")
	}
	return true
}
func sendStartMsgInit() *mq.MonitorMsgMq {
	var wg sync.WaitGroup
	baseMsgSendBody = mq.NewMonitorMsgMq()
	baseMsgSendBody.SendMsg.MsgModifyKey = PlanContext.ChanSchedulePlanTable
	baseMsgSendBody.NormalQuit = global.EFunc{FuncName: sendStartMsgFinishFunc}
	baseMsgSendBody.ContinueFunc = global.EFunc{FuncName: sendStartMsgContinueFunc}
	baseMsgSendBody.ExecFunc = append(baseMsgSendBody.ExecFunc, global.EFunc{FuncName: sendStartMsgExecFunc})
	baseMsgSendBody.CurryLimit = mq.SendCurryLimit{
		Switch:   true,
		Wg:       &wg,
		CurrySum: make(chan struct{}, GlobalPConfigs.rules.Mtc),
	}
	return baseMsgSendBody
}

func schedulePlanStart1() {
	if PlanContext.Pool.Submit(func() {
		//发送消息
		fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task sync -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		global.MyWait.Add(1)
		sendStartMsgInit().NormalChanSendMsg("ckR/schedulePlanStart1")
		fmt.Println(fmt.Sprintf("-- [%v] %v report: schedule task sync finish -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	}) != nil {
		return
	}
}

func (context *SchedulePlanContext) closeMonitor() {
	context.Close()
	//关闭消费者 阻塞
	consumer.Shutdown()
	//关闭生产者
	producer.Shutdown()
	newTer.Stop()
	newCell.Stop()
}
func (context *SchedulePlanContext) endClose() {
	if err := xlsFile.Close(); err != nil {
		return
	}
	closeObject.CloseAction()
	fmt.Println("")
	//输出结果信息
	fmt.Println("Check end")
	infoWriteInfo(fmt.Sprintf("%v check object {%v} complete !!!", arg.ToolName, GlobalPConfigs.rules.CheckMode))
	fmt.Println(fmt.Sprintf("** %v Overview of results **", arg.ToolName))
	fmt.Println("Check time: ", fmt.Sprintf("%.2fs", time.Since(context.BeginTime).Seconds()), "(Seconds)")
	if err := newTer.Load("execTime", fmt.Sprintf("%v", time.Since(context.BeginTime).Seconds())); err != nil {
		errorWriteInfo(fmt.Sprintf("%v", err))
	}
	newTer.Print()
}
func (context *SchedulePlanContext) Start(wg *sync.WaitGroup) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task monitor -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	terminal(wg)
	context.monitor()
	Cell(wg)
}
func (context *SchedulePlanContext) Check() {
	var wg sync.WaitGroup
	context.Start(&wg)
	closeObject.AbnormalExit()
	//执行计划开始
	schedulePlanStart1()
	global.MyWait.Wait()
	//关闭检测
	context.closeMonitor()
	wg.Wait()
	context.endClose()
}

// Audit input
func Audit(m arg.ConfigParameter) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin input checkData mode -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	if m.ArgInput.OptimizerSwitch.CpuOptimizer {
		defer pprof.StopCPUProfile()
		ref.CpuAnalyze()
	}
	if m.ArgInput.OptimizerSwitch.MemOptimizer {
		f := ref.MemAnalyze()
		defer func() {
			if err := f.Close(); err != nil {
				return
			}
		}()
	}
	ref.MyWaitInit()
	schemaContext(&SchemaInitContextConfig{
		GtCheckConfig: m,
	}).Check()
	schedulePlanContext(&SchedulePlanContextConfig{
		GtCheckConfig: m,
		MQConfig:      cp.NewDefaultMq(),
		BeginTime:     m.BeginTime,
	}).Check()
}
